package com.cskaoyan.mr.topn;


import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.util.StringTokenizer;
import java.util.TreeSet;

/**
 * @ClassName TopNMapper
 * @Description:
 * KEYIN:当前行相当于整篇内容的一个偏移量；一般情况下不会去获取
 * VALUEIN：当前传入进行的一行数据
 * KEYOUT：map函数运行完毕之后，输出的key值
 * VALUEOUT:map函数运行完毕之后，输出的value值
 * @Author 远志 zhangsong@cskaoyan.onaliyun.com
 * @Date 2023/2/13 14:18
 * @Version V1.0
 **/
public class TopNMapper extends Mapper<Object, Text, IntWritable, NullWritable> {

    private TreeSet<Integer> treeSet = new TreeSet<>();
    /**
     * 每当处理一行数据，那么便会调用一个map函数
     * 823 2378327 237823
     * 823982 387374 8933
     * 833748 347834 3849
     * 283 2873 2938 38
     * 283 2893 3489 2389
     * 283 389 2389 3489
     *
     * 12 23 35
     * 5 23 35
     * @param key
     * @param value
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
        //这里面调用哪个方法，取决于input输入里面的文件格式是什么样的
        //如果是以空格分割的数字，那么我们就采取如下这种方式
        //如果每一行就只有唯一的一个数字，那么反而更简单了
        StringTokenizer stringTokenizer = new StringTokenizer(value.toString());
        while (stringTokenizer.hasMoreTokens()){
            //这里面的character就是数字
            String character = stringTokenizer.nextToken();
            treeSet.add(Integer.parseInt(character));
            if(treeSet.size() > TopN.N){
                treeSet.pollFirst();
            }
        }
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        //该方法会在当前节点的所有行数据处理完毕之后，调用一次cleanup
        for (Integer integer : treeSet) {
            context.write(new IntWritable(integer), NullWritable.get());
        }
    }
}
